package com.intct.utils;

import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import com.ververica.cdc.connectors.mysql.table.StartupOptions;
import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;
import org.apache.flink.api.java.utils.ParameterTool;

/**
 * @author gufg
 * @since 2025-09-17 16:42
 */
public class MysqlUtil {
    /**
     * 配置MySQL源
     * @param parameterTool
     * @return
     */
    public static MySqlSource<String> getMysqlSource(ParameterTool parameterTool, String tableList, StartupOptions startupOptions) {
        return MySqlSource.<String>builder()
                .hostname(parameterTool.get("mysql.host", "node1"))
                .username(parameterTool.get("mysql.user.name", "root"))
                .password(parameterTool.get("mysql.password", "123456"))
                .port(parameterTool.getInt("mysql.port", 3306))
                .databaseList(parameterTool.get("mysql.database", "travel"))
                .tableList(tableList)
                .deserializer(new JsonDebeziumDeserializationSchema())
                .startupOptions(startupOptions)
                .build();
    }

    /**
     * 配置MySQL源
     * @param parameterTool
     * @return
     */
    public static MySqlSource<String> getMysqlSource(ParameterTool parameterTool, String dbList, String tableList, StartupOptions startupOptions) {
        return MySqlSource.<String>builder()
                .hostname(parameterTool.get("mysql.host", "node1"))
                .username(parameterTool.get("mysql.user.name", "root"))
                .password(parameterTool.get("mysql.password", "123456"))
                .port(parameterTool.getInt("mysql.port", 3306))
                .databaseList(dbList)
                .tableList(tableList)
                .deserializer(new JsonDebeziumDeserializationSchema())
                .startupOptions(startupOptions)
                .build();
    }
}
